package com.atguigu.gmallrealtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmallrealtime.bean.TableProcess;
import com.atguigu.gmallrealtime.util.JDBCUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.HashMap;
import java.util.List;

/**
 * @author yhm
 * @create 2023-09-23 15:27
 */
public class DimConnectStreamBroadcastFunc  extends BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<TableProcess, JSONObject>>  {
    private HashMap<String, TableProcess> hashMap = new HashMap<String, TableProcess>();

    private MapStateDescriptor<String, TableProcess> tableProcessDesc ;
    public DimConnectStreamBroadcastFunc(MapStateDescriptor<String, TableProcess> tableProcessDesc) {
        this.tableProcessDesc=tableProcessDesc;
    }


    /**
     * 主流数据可能比配置流数据到的早   造成错误的数据删除  所以先加载一份配置表数据
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        // 1. 读取配置表
        Connection connection = JDBCUtil.getMysqlConnection();

        // 2. 存储到本地
        try {
            List<TableProcess> tableProcesses = JDBCUtil.queryList(connection, "select * from gmall_config.table_process", TableProcess.class, true);
            for (TableProcess tableProcess : tableProcesses) {
                tableProcess.setOp("init");
                hashMap.put(tableProcess.getSourceTable() + tableProcess.getSourceType(),tableProcess);
            }
        }catch (Exception e){
            System.out.println("初始化加载配置表报错");
        }

        // 3. 关闭连接 释放资源
        connection.close();
    }

    /**
     *  处理主流数据  处理两种数据不分时间先后  谁先来谁先算
     * @param jsonObj
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<Tuple2<TableProcess, JSONObject>> out) throws Exception {
        // 1. 获取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessDesc);

        // 2. 判断当前数据属于的表格是否存在于状态中
        String tableName = jsonObj.getString("table");
        String sourceType = "ALL";
        String key = tableName + sourceType;
        TableProcess dimState = broadcastState.get(key);

        if (dimState == null){
            // 说明状态中没有这个表
            dimState = hashMap.get(key);
        }

        if (dimState != null) {
            // 3. 过滤出存在状态中的维度表 读取出tableProcess
            // 4. 拼接写出
            TableProcess tableProcess = broadcastState.get(key);
            if (tableProcess == null){
                tableProcess = hashMap.get(key);
            }
            out.collect(Tuple2.of(tableProcess, jsonObj));

        } else {
            // 当前数据不是维度表数据  不需要
        }
    }

    /**
     * 处理广播流数据
     * @param tp
     * @param ctx
     * @param out
     * @throws Exception
     */
    @Override
    public void processBroadcastElement(TableProcess tp, Context ctx, Collector<Tuple2<TableProcess, JSONObject>> out) throws Exception {
        // 1. 判断当前配置表数据的类型
        // crud
        String op = tp.getOp();
        // 2. 获取状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(tableProcessDesc);
        // 3. 根据当前数据的类型 对状态进行修改
        String key = tp.getSourceTable() + tp.getSourceType();
        if ("d".equals(op)) {

            broadcastState.remove(key);
            // map中也对应删除
            hashMap.remove(key);
        } else {
            broadcastState.put(key, tp);

        }

    }
}
